package io.github.wanggit.antrpc.client.zk.listener;

import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.client.zk.register.IZkRegisterHolder;
import io.github.wanggit.antrpc.client.zk.zknode.INodeHostContainer;
import io.github.wanggit.antrpc.client.zk.zknode.IReportSubscriber;
import io.github.wanggit.antrpc.client.zk.zknode.NodeHostEntity;
import io.github.wanggit.antrpc.client.zk.zknode.NodeHostEntityHelper;
import io.github.wanggit.antrpc.commons.bean.InterfaceNodeDataBean;
import io.github.wanggit.antrpc.commons.bean.InterfaceNodeDataBeanHelper;
import io.github.wanggit.antrpc.commons.bean.IpNodeDataBean;
import io.github.wanggit.antrpc.commons.bean.RegisterBeanMethod;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.rate.IResilience4jRateLimiter;
import io.github.wanggit.antrpc.commons.utils.ApplicationNameUtil;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.ZkNodeType;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;

import java.util.Map;
import java.util.Objects;

/**
 *
 *
 * <pre>
 *     监听 /__antrpc_na__ 节点的变化, 服务消费者与服务提供者本身都会监听此变化。
 *     服务消费者从中得知可用接口列表；服务提供者从中得知接口服务提供端限流信息；
 * </pre>
 */
@Slf4j
public class RegisterNodeChangeListener implements IListener {

    private final IZkClient zkClient;
    private final IZkRegisterHolder zkRegisterHolder;
    private final INodeHostContainer nodeHostContainer;
    private final IConfiguration configuration;
    private final IResilience4jRateLimiter serverResilience4jRateLimiterContainer;
    private final IReportSubscriber reportSubscriber;

    public RegisterNodeChangeListener(
            IZkClient zkClient,
            IZkRegisterHolder zkRegisterHolder,
            INodeHostContainer nodeHostContainer,
            IConfiguration configuration,
            IResilience4jRateLimiter serverResilience4jRateLimiterContainer,
            IReportSubscriber reportSubscriber) {
        this.zkClient = zkClient;
        this.zkRegisterHolder = zkRegisterHolder;
        this.nodeHostContainer = nodeHostContainer;
        this.configuration = configuration;
        this.serverResilience4jRateLimiterContainer = serverResilience4jRateLimiterContainer;
        this.reportSubscriber = reportSubscriber;
    }

    @Override
    public void listen() {
        internalListen(zkClient.getCurator());
    }

    /**
     *
     *
     * <pre>
     *     <[CDATA[
     *     /__rpc_na__/172.16.101.19:6060/io.github.wanggit.antrpc.demo.api.RateLimitService
     * {
     *   "ts" : 1593486028212,
     *   "methods" : [ "trim(java.lang.String)", "lowerIt(java.lang.String)", "upperIt(java.lang.String)" ],
     *   "methodMap" : {
     *     "lowerIt(java.lang.String)" : {
     *       "className" : "io.github.wanggit.antrpc.demo.rate_limit_provider.service.RateLimitServiceImpl",
     *       "methodName" : "lowerIt",
     *       "parameterTypeNames" : [ "java.lang.String" ],
     *       "limit" : 10,
     *       "durationInSeconds" : 2,
     *       "methodId" : null,
     *       "humanFullName" : null
     *     },
     *     "trim(java.lang.String)" : {
     *       "className" : "io.github.wanggit.antrpc.demo.rate_limit_provider.service.RateLimitServiceImpl",
     *       "methodName" : "trim",
     *       "parameterTypeNames" : [ "java.lang.String" ],
     *       "limit" : 0,
     *       "durationInSeconds" : 0,
     *       "methodId" : null,
     *       "humanFullName" : null
     *     },
     *     "upperIt(java.lang.String)" : {
     *       "className" : "io.github.wanggit.antrpc.demo.api.RateLimitService",
     *       "methodName" : "upperIt",
     *       "parameterTypeNames" : [ "java.lang.String" ],
     *       "limit" : 10,
     *       "durationInSeconds" : 2,
     *       "methodId" : null,
     *       "humanFullName" : null
     *     }
     *   },
     *   "className" : "io.github.wanggit.antrpc.demo.api.RateLimitService",
     *   "hostInfo" : "172.16.101.19:6060",
     *   "fullPath" : "/__rpc_na__/172.16.101.19:6060/io.github.wanggit.antrpc.demo.api.RateLimitService",
     *   "appName" : "rate_limit_provider"
     * }
     *
     * methodMap => key -> RegisterBeanMethod.toString, value -> registerBeanMethod
     *     ]]>
     * </pre>
     *
     * 监听节点接口注册的变化
     *
     * @param curator curator
     */
    private void internalListen(CuratorFramework curator) {
        TreeCache rootNodeTreeCache =
                new TreeCache(curator, "/" + ConstantValues.ZK_ROOT_NODE_NAME);
        rootNodeTreeCache
                .getListenable()
                .addListener(
                        (client, event) -> {
                            if (TreeCacheEvent.Type.NODE_ADDED.equals(event.getType())) {
                                doNodeAddedEvent(event);
                            } else if (TreeCacheEvent.Type.NODE_UPDATED.equals(event.getType())) {
                                doNodeUpdatedEvent(event);
                            } else if (TreeCacheEvent.Type.NODE_REMOVED.equals(event.getType())) {
                                doNodeRemovedEvent(event);
                            } else if (TreeCacheEvent.Type.CONNECTION_RECONNECTED.equals(
                                    event.getType())) {
                                if (log.isInfoEnabled()) {
                                    log.info("Zookeeper reconnects and re-registers the service.");
                                }
                                zkRegisterHolder.allReRegister();
                            }
                        });
        try {
            rootNodeTreeCache.start();
        } catch (Exception e) {
            String errorMessage = "TreeCache failed to start.";
            if (log.isErrorEnabled()) {
                log.error(errorMessage, e);
            }
            throw new ZkListenerException(errorMessage, e);
        }
    }

    private void doNodeRemovedEvent(TreeCacheEvent event) {
        ChildData data = event.getData();
        String path = data.getPath();
        if (log.isInfoEnabled()) {
            log.info("The " + data.getPath() + " node has been deleted.");
        }
        ZkNodeType.RegisterNodeType registerNodeType =
                ZkNodeType.getRegisterNodeType(data.getPath());
        if (Objects.equals(registerNodeType, ZkNodeType.RegisterNodeType.IP)) {
            IpNodeDataBean ipNodeDataBean =
                    JsonUtils.fromJson(data.getData(), IpNodeDataBean.class);
            if (null != ipNodeDataBean) {
                String hostInfo =
                        ApplicationNameUtil.getHost(ipNodeDataBean.getAppName()).getHostInfo();
                nodeHostContainer.deleteIpAndAppNamePair(hostInfo);
            }
        } else if (Objects.equals(registerNodeType, ZkNodeType.RegisterNodeType.INTERFACE)) {
            InterfaceNodeDataBean nodeData =
                    InterfaceNodeDataBeanHelper.getNodeDataWithoutRateLimiting(data.getData());
            if (wasSubscribed(nodeData)) {
                NodeHostEntity hostEntity =
                        NodeHostEntityHelper.createNodeHostEntity(path, nodeData);
                nodeHostContainer.delete(hostEntity.getClassName(), hostEntity);
            }
        }
    }

    private void doNodeUpdatedEvent(TreeCacheEvent event) {
        ChildData data = event.getData();
        String path = data.getPath();
        if (log.isInfoEnabled()) {
            log.info("The " + path + " node data has been updated.");
        }
        internalDoNodeAddedOrUpdatedEvent(data, path, false);
    }

    private void doNodeAddedEvent(TreeCacheEvent event) {
        ChildData data = event.getData();
        String path = data.getPath();
        internalDoNodeAddedOrUpdatedEvent(data, path, true);
    }

    private void internalDoNodeAddedOrUpdatedEvent(ChildData data, String path, boolean added) {
        ZkNodeType.RegisterNodeType registerNodeType =
                ZkNodeType.getRegisterNodeType(data.getPath());
        if (Objects.equals(registerNodeType, ZkNodeType.RegisterNodeType.IP)) {
            doIpNodeAddedOrUpdatedEvent(data);
        } else if (Objects.equals(registerNodeType, ZkNodeType.RegisterNodeType.INTERFACE)) {
            doInterfaceNodeAddedOrUpdatedEvent(data, path, added);
        }
    }

    private void doInterfaceNodeAddedOrUpdatedEvent(ChildData data, String path, boolean added) {
        InterfaceNodeDataBean nodeData =
                InterfaceNodeDataBeanHelper.getNodeDataWithoutRateLimiting(data.getData());
        // 如果当前节点订阅了此接口，那么可以加入到本地缓存中
        if (wasSubscribed(nodeData)) {
            if (log.isInfoEnabled()) {
                log.info(path + " node save or update to local cache.");
            }
            NodeHostEntity hostEntity = NodeHostEntityHelper.createNodeHostEntity(path, nodeData);
            if (added) {
                nodeHostContainer.add(hostEntity.getClassName(), hostEntity);
            } else {
                nodeHostContainer.update(hostEntity.getClassName(), hostEntity);
            }
        }
        // 如果是当前节点注册的此接口，那么处理可能存在的服务提供者端限流
        if (isCurrentNode(nodeData)) {
            checkOrUpdateCurrentNodeServerRateLimitInfo(data.getData());
        }
    }

    private void doIpNodeAddedOrUpdatedEvent(ChildData data) {
        IpNodeDataBean ipNodeDataBean = JsonUtils.fromJson(data.getData(), IpNodeDataBean.class);
        if (null != ipNodeDataBean) {
            String hostInfo =
                    ApplicationNameUtil.getHost(ipNodeDataBean.getAppName()).getHostInfo();
            String simpleAppName =
                    ApplicationNameUtil.getSimpleAppName(ipNodeDataBean.getAppName());
            nodeHostContainer.addOrUpdateIpAndAppNamePair(hostInfo, simpleAppName);
        }
    }

    /**
     * 每个节点当监听到本身的接口注册时，从接口节点数据中获取限流信息，并在本节点内处理
     *
     * @param bytes data
     */
    private void checkOrUpdateCurrentNodeServerRateLimitInfo(byte[] bytes) {
        InterfaceNodeDataBean nodeData = JsonUtils.fromJson(bytes, InterfaceNodeDataBean.class);
        if (log.isInfoEnabled()) {
            log.info("Current node update limit info.");
        }
        Map<String, RegisterBeanMethod> methodMap = nodeData.getMethodMap();
        if (null != methodMap) {
            methodMap
                    .values()
                    .forEach(
                            it -> {
                                // RegisterBeanMethod中的class可能是实现类，把它替换为接口
                                it.setClassName(nodeData.getClassName());
                                serverResilience4jRateLimiterContainer
                                        .checkOrChangeEventCountCircuitBreaker(it);
                            });
        }
    }

    private boolean isCurrentNode(InterfaceNodeDataBean nodeDataBean) {
        return Objects.equals(
                nodeDataBean.getHostInfo(),
                configuration.getExposeIp() + ":" + configuration.getPort());
    }

    private boolean wasSubscribed(InterfaceNodeDataBean nodeDataBean) {
        return reportSubscriber.wasSubscribed(nodeDataBean.getClassName());
    }
}
